/*
*Desc:	Uginxd that reference as Nginx + Memcached(Nginx = Master process + worker process, Memcached = main thread + working threads)
*		the Uginxd will receive request from client and pack the request, 
*			after packing, request will be dispatched to process thread which execute process logic with engine.
*		Uginxd impl with libevent.The process of engine can be run with async and sync.
*		master worker structure is useful, the master create worker processes(reference as nginx) and the worker process will create specific working threads.
*		when setup, suggest that process=1 and threads = cups
*
*Author:clouduser@163.com, clouduser.moo@gmail.com
*
*CreateDate: 2014-05
*/

#include <sys/stat.h>
#include <signal.h>
#include <sys/resource.h>
#include <ctype.h>
#include <stdarg.h>

#include <unistd.h>
#include <getopt.h>
#include <pthread.h>
/* some POSIX systems need the following definition
 * to get mlockall flags out of sys/mman.h.  */
#ifndef _P1003_1B_VISIBLE
#define _P1003_1B_VISIBLE
#endif
/* need this to get IOV_MAX on some platforms. */
#ifndef __need_IOV_MAX
#define __need_IOV_MAX
#endif

#include <pwd.h>
#include <sys/mman.h>
#include <fcntl.h>
#include <netinet/tcp.h>
#include <arpa/inet.h>
#include <errno.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <time.h>
#include <assert.h>
#include <limits.h>
#include <sysexits.h>
#include <stddef.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <ifaddrs.h>
#include <net/if.h>

#ifdef HAVE_DLFCN_H
#include <dlfcn.h>
#endif

#ifdef HAVE_LINK_H
#include <link.h>
#endif

#include <sys/ioctl.h>

#include <list>
#include <vector>
#include <sstream>
#include <tr1/functional>

////////////
#include "uginx.h"

#include "net/net.h"
#include "net/conn.h"
#include "threads/thread.h"
#include "threads/thread_key.h"
#include "engine/engine.h"
#include "utility/ugx_palloc.h"
#include "master_workers/mw_process_cycle.h"
#include "utility/ini_parser.h"

#include <glog/logging.h>
#include <gflags/gflags.h>

///////////
using namespace std;

DEFINE_int32(port, 6699,"port");
DEFINE_string(iplist, "127.0.0.1","localhost");
DEFINE_int32(num_thread, 8,"num_thread_per_process");
DEFINE_int32(num_process, 1,"num_process");
DEFINE_bool(daemon, false, "daemonize");
DEFINE_bool(maxcore, false, "set max corefile");
DEFINE_string(soname, "","soname that handle specific request and ret response");
DEFINE_string(engine_args, "","after loading soname, need give args for engine");
DEFINE_string(master_pid, "","master pid file");
DEFINE_bool(enable_async, false, "enable async process request");
DEFINE_string(config, "../config/config.ini","config.ini");

/*config ini map*/
std::map<std::string, std::string>__config;



//pid_t    ps_pid ;
typedef void (*spawn_proc_pt) (void *data);
typedef void (*event_handler_pt)();

struct settings settings;
extern pthread_key_t tsd_key;
struct event conn_event;  
volatile int recv_quit_flag;
pthread_t *recv_tids;
int recv_num_threads;

static void settings_init(void) 
{
	settings.listenfd = -1;
	settings.pid_file_fd = -1;
	settings.queue_lwm = 1;
	settings.queue_hwm = 1024;
	settings.engine_argv = NULL;
	settings.engine_argc = 1;
	//settings.num_cpus = sysconf(_SC_NPROCESSORS_CONF);
	settings.port = FLAGS_port;
	settings.num_threads_per_process = FLAGS_num_thread;
	settings.num_processes = FLAGS_num_process;
	settings.do_daemonize = FLAGS_daemon;
	settings.engine_soname = const_cast<char*>(FLAGS_soname.c_str());
	settings.engine_args = const_cast<char*>(FLAGS_engine_args.c_str());
	settings.master_pid_file = const_cast<char*>(FLAGS_master_pid.c_str());
	settings.enable_async_process = FLAGS_enable_async;

	if (NULL != settings.engine_args) {
		settings.engine_argc = 1;
		int x=1;//x=split count
		settings.engine_argv = (char **)malloc(x*sizeof(char*));
		for(int i=0;i<x;++i){
			;//settings.engine_argv[i]; //split[i]
		}
	}
}

#if (!defined(HAVE_SIGIGNORE) || defined(__FreeBSD__) )
static int sigignore(int sig) {
	struct sigaction sa = { {SIG_IGN}, 0 };
	if (sigemptyset(&sa.sa_mask) == -1 || sigaction(sig, &sa, 0) == -1) {
		return -1;
	}
	return 0;
}
#endif


static int process_async_callback(void* result, size_t result_len, void* user_ptr)
{
	request_t* request = (request_t*)user_ptr;
	if (request == NULL) {
		LOG(WARNING) << "socket fd lost!!!,[process_async_callback] this will expense a fd";
		return -1;
	}

	if (result == NULL || result_len == 0) {
		request_free(request);
		return 0;
	}

	if (result_len > 4096) {
		request_free(request);
		return -1;
	}

	//process_request(request);

	request_free(request);
	return 0;
}

static void process_queue(struct event_base* evbase, void* data, void* userdata)
{
	request_t* request = (request_t*)data;
	if (request == NULL){
		LOG(WARNING) << "socket fd lost!!!,[process_queue] this will expense a fd";
		return;
	}

	_pool_t* pool = _create_pool(UGX_MAX_ALLOC_FROM_POOL, UGX_MAX_ALLOC_FROM_POOL); 
	if (pool == NULL) {
		LOG(WARNING) << "_create_pool failed!";
		request_free(request);
		return;
	}

	request->pool = pool;
	request->evbase = evbase;

	if (settings.enable_async_process) {
		process_async(request->pool, request->buffer, request->buffer_size, process_async_callback, request);
	} else {
		size_t   response_size = 0; 
		uint8_t*response = process_sync(request->pool, request->buffer, request->buffer_size, request, &response_size);
		request_free(request);
	}
}


static void worker_exit(void* data)
{
	thread_destroy();
	unload_engine();
	net_interface_destroy(settings.interfaces);
	settings.interfaces.clear();
	free(settings.engine_argv);
	thread_key_delete();
}

void worker_main(struct event_base* base, void* data)
{
	if (thread_key_create() != 0) {
		LOG(WARNING) << "init thread_key failed.";
		exit(EXIT_FAILURE);
	}
	/*init LIBEVENT_THREAD struct, and set circle queue*/
	thread_init(settings.num_threads_per_process, settings.queue_hwm, settings.queue_lwm, process_queue, NULL);

	// set connection_accept event
	event_set(&conn_event, settings.listenfd, EV_READ | EV_PERSIST, connection_accept, base);
	event_base_set(base, &conn_event);
	event_add(&conn_event, NULL);
}

void set_max_core() 
{
	struct rlimit rlim;
	struct rlimit rlim_new;
	/*
	 * First try raising to infinity; if that fails, try bringing
	 * the soft limit to the hard.
	 */
	if (getrlimit(RLIMIT_CORE, &rlim) == 0) {
		rlim_new.rlim_cur = rlim_new.rlim_max = RLIM_INFINITY;
		if (setrlimit(RLIMIT_CORE, &rlim_new)!= 0) {
			/* failed. try raising just to the old max */
			rlim_new.rlim_cur = rlim_new.rlim_max = rlim.rlim_max;
			(void)setrlimit(RLIMIT_CORE, &rlim_new);
		}
	}
	/*
	 * getrlimit again to see what we ended up with. Only fail if
	 * the soft limit ends up 0, because then no core files will be
	 * created at all.
	 */
	if ((getrlimit(RLIMIT_CORE, &rlim) != 0) || rlim.rlim_cur == 0) {
		fprintf(stderr, "failed to ensure corefile creation\n");
		exit(EX_OSERR);
	}
}

void set_daemon()
{
	/* daemonize if requested */
	/* if we want to ensure our ability to dump core, don't chdir to / */
		if (settings.master_pid_file == NULL) {    
			fprintf(stderr, "Error: --pid-file or -P is not specified.\n");
			exit(-1);
		} else {    
			if (settings.master_pid_file[0] != '/') {    
				fprintf(stderr, "Error: pid-file '%s' is not a valid absolute path.\n", settings.master_pid_file);
				exit(-1);
			}    
		}    

		if (daemon(FLAGS_maxcore, 0) == -1) {
			fprintf(stderr, "failed to daemon() in order to daemonize\n");
			exit(EXIT_FAILURE);
		}
		if ((settings.pid_file_fd = checkonly(settings.master_pid_file)) < 0) {
			fprintf(stderr, "Error: Process already running.\n");
			exit(EXIT_FAILURE);
		}
}

/*
 *main()
 *params: FLAGS setting
 *output: none
 */
int main (int argc, char **argv) 
{
	/*initialize config*/
	INIParser __ini(FLAGS_config.c_str());//iniparser
	int iniok = __ini.parser();
	if (0 != iniok){
		printf("iniparser fail");
		exit(EXIT_FAILURE);
	}
	__config = __ini.getKeyValBySection("uginx");

	//glog
	::google::ParseCommandLineFlags(&argc, &argv, true);
	::google::SetLogDestination(::google::GLOG_ERROR, (__config["logpath"] + "error.").c_str());
	::google::SetLogDestination(::google::FATAL, (__config["logpath"]+"fatal.").c_str());
	::google::SetLogDestination(::google::INFO, (__config["logpath"]+"info.").c_str());
	::google::SetLogDestination(::google::WARNING, (__config["logpath"]+"warning.").c_str());
	google::InitGoogleLogging(argv[0]);

	/* init settings that init FLAGS_*        */
	settings_init();

	/* check input parameters */
	if (FLAGS_num_process <= 0 || FLAGS_num_thread <= 0 ){
		printf("num thread or process <=0\n");
		printf("%d \t %d\n", settings.num_processes, settings.num_threads_per_process);
		exit(EXIT_FAILURE);
	}
	if ( FLAGS_soname == ""){
		printf("soname not set\n");
		exit(EXIT_FAILURE);
	}else{
		if ( FLAGS_soname[0] != '/' || access(FLAGS_soname.c_str(),F_OK) != 0 ){
			printf("soname not absolute path or soname file not exist\n");
			exit(EXIT_FAILURE);
		}
	}
	if ( sigignore(SIGPIPE) == -1 ){
		printf("signal ignore failed!\n");
		exit(EXIT_FAILURE);
	}

	/* set stderr non-buffering*/
	setbuf(stderr, NULL);

	/* set max corefile*/
	if (FLAGS_maxcore){
		set_max_core();
	}

	/*create listen socket and set nonblock listenfd*/
	if (FLAGS_port > 0) {
		settings.listenfd = create_socket(NULL, FLAGS_port, SOCK_STREAM);
		int ret_listen = setnonblock(settings.listenfd);
		if (0 != ret_listen) {
			printf("failed to setnonblock on listenfd");
			exit(EX_OSERR);
		}
		if (settings.listenfd == -1) {
			printf("failed to listen on port:%d\n",settings.port);
			exit(EX_OSERR);
		}
	}

	/*load engine that include all processing functions*/
	if (!load_engine(settings.engine_soname, settings.engine_argc, settings.engine_argv)) {
		printf("load_engine failed\n");
		exit(EXIT_FAILURE);
	}

	/* set proc daemonize*/
	if (FLAGS_daemon) {
		set_daemon();
	}

	mw_master_process_cycle(settings.num_processes, worker_main, worker_exit, NULL);

	google::ShutdownGoogleLogging();
	return EXIT_SUCCESS;
}
